package org.witsoft.localmq.impl;

import org.witsoft.localmq.ConfigManager;
import org.witsoft.localmq.FileManager;
import org.witsoft.localmq.OffsetManager;
import org.witsoft.localmq.model.Offset;
import org.witsoft.localmq.model.Segment;
import org.witsoft.localmq.model.SegmentList;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.Synchronized;
import lombok.extern.slf4j.Slf4j;

import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.channels.FileLock;
import java.util.Collection;
import java.util.List;

/**
 * @ClassName: MqOffsetManagerImpl
 * @Description:
 * @Author: Colin
 * @Date: 2020/12/26 11:16 AM
 * @Version 1.0
 **/
@Slf4j
@RequiredArgsConstructor
public class OffsetManagerImpl implements OffsetManager {

    @NonNull
    private ConfigManager configManager;

    @NonNull
    private FileManager fileManager;

    private SegmentList segmentList;

    private long produceOffset = 0;

    private long consumeOffset = 0;

    private volatile RandomAccessFile indexFile;

    private FileLock indexFileLock;

    @Override
    public long getProduceOffset() {
        try {
            checkLoadIndexData();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return produceOffset;
    }

    @Synchronized
    public Offset nextProduceOffset(long dataSize) throws IOException{
        try {
            checkLoadIndexData();
        } catch (IOException e) {
           throw new RuntimeException(e);
        }
        Offset offset ;
        Segment segment = segmentList.getFirst();
        if(segment == null){
            offset = Offset.builder()
                    .segmentId(1)
                    .position(0)
                    .startOffset(0)
                    .currentOffset(0)
                    .build();
        }else{
            // 是否超出限制大小
            long maxSize = configManager.getMaxSegmentFileSize();
            if(segment.getFileSize() + dataSize > maxSize){
                offset = Offset.builder()
                        .segmentId(segment.getId() + 1)
                        .position(0)
                        .startOffset(segment.getEndOffset() + 1)
                        .currentOffset(segment.getEndOffset() + 1)
                        .build();
            }else{
                offset = Offset.builder()
                        .segmentId(segment.getId())
                        .position(segment.getPosition() + 1)
                        .startOffset(segment.getStartOffset())
                        .currentOffset(segment.getEndOffset() + 1)
                        .build();
            }
        }
        // 保存产生的偏移量
        this.commitProduceOffset(offset, dataSize);
        // 返回偏移量
        return offset;
    }

    @Override
    public long getConsumeOffset() {
        try {
            checkLoadIndexData();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return consumeOffset;
    }

    @Override
    public void commitConsumeOffset(long num) throws IOException {
        if(num > this.produceOffset + 1){
//            throw new IOException("消费偏移量大于生产偏移量");
            return;
        }
        RandomAccessFile raf = this.indexFile;
        raf.seek(8);
        raf.writeLong(num);
        this.consumeOffset = num;
    }

    @Override
    public Collection<Segment> getSegments() {
        return segmentList.readOnly();
    }

    @Override
    @Synchronized
    public void purge () throws IOException{
        // 过滤比当前消费的偏移量小的片段
        List<Segment> segments = segmentList.filterEndOffsetLessThan(consumeOffset);
        segments.sort((a,b)->Integer.valueOf(a.getId()).compareTo(b.getId()));
        //保留最后一个
        if(segments.size() > 0) {
            segments.remove(segments.size() - 1);
        }
        // 删除分片索引和数据文件
        for (Segment segment : segments) {
            File idxFile = fileManager.getSegmentDataFile(segment.getId());
            File dataFile = fileManager.getSegmentIndexFile(segment.getId());
            log.info("准备清理片段文件:{}", dataFile.getName());
            if(idxFile.delete()){
                log.info("成功清理文件:{}", idxFile.getPath());
            }
            if(dataFile.delete()){
                log.info("成功清理文件:{}", dataFile.getPath());
            }
        }
        if(segments.isEmpty()){
            log.info("暂无可清理的分段文件!");
        }
        // 重新保存索引
        RandomAccessFile raf = this.indexFile;
        for (Segment segment : segments) {
            // 写到当前索引
            raf.seek(16 + 24 * (segment.getId() - 1));
            raf.writeInt(-1);
            raf.writeLong(-1);
            raf.writeInt(-1);
            raf.writeLong(0);
            // 移除索引
            segmentList.remove(segment);
        }
    }

    public Offset getOffset(long num){
        try {
            checkLoadIndexData();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        Segment segment = segmentList.match(num);
        if(segment == null) {
            return null;
        }
        Offset offset = Offset.builder()
                .segmentId(segment.getId())
                .startOffset(segment.getStartOffset())
                .position((int)(num - segment.getStartOffset()))
                .currentOffset(num)
                .build();
        return offset;
    }

    private void checkLoadIndexData() throws IOException{
        if(this.indexFile != null){
            return;
        }
        synchronized (this){
            if(this.indexFile == null) {
                File file = fileManager.getIndexFile();
                RandomAccessFile raf = new RandomAccessFile(file, "rw");
                this.loadIndexData(raf);
                this.indexFile = raf;
                this.indexFileLock = raf.getChannel().lock();
                if(this.indexFileLock.isValid()){
                    log.debug("文件[{}]已上锁", file.getPath());
                }
            }
        }
    }

    private void commitProduceOffset(Offset logOffset , long dataSize) throws IOException{
        Segment newSegment = null;
        Segment segment = segmentList.findById(logOffset.getSegmentId());
        if(segment == null){
            newSegment = new Segment();
            newSegment.setId(logOffset.getSegmentId());
            newSegment.setStartOffset(logOffset.getStartOffset());
            newSegment.setPosition(0);
            newSegment.setFileSize(0);
            segment = newSegment;
        }
        segment.setPosition(logOffset.getPosition());
        segment.setFileSize(segment.getFileSize() + dataSize);
        // 记录新的偏移量
        long newOffset = segment.getEndOffset();
        RandomAccessFile raf = this.indexFile;
        raf.seek(0);
        raf.writeLong(newOffset);
        produceOffset = newOffset;
        // 写到当前索引
        raf.seek(16 + 24 * (segment.getId() - 1));
        raf.writeInt(segment.getId());
        raf.writeLong(segment.getStartOffset());
        raf.writeInt(segment.getPosition());
        raf.writeLong(segment.getFileSize());
        // 加入到集合中
        if(segment == newSegment) {
            segmentList.addFirst(newSegment);
        }
    }

    private void loadIndexData(RandomAccessFile raf) throws IOException{
        if(raf.length() == 0){
            segmentList = new SegmentList();
            return;
        }
        // 读取当前偏移量
        raf.seek(0);
        produceOffset = raf.readLong();
        consumeOffset = raf.readLong();
        // 读取片段
        segmentList = new SegmentList();
        try{
            while(true){
                long pointer = raf.getFilePointer();
                if(pointer >= raf.length()){
                    break;
                }
                int id = raf.readInt();
                Segment segment = new Segment();
                segment.setId(id); // 4
                segment.setStartOffset(raf.readLong()); // 8
                segment.setPosition(raf.readInt()); // 4
                segment.setFileSize(raf.readLong()); // 8
                if(id == -1){
                    continue;
                }
                segmentList.addFirst(segment);
            }
        }catch (EOFException e){
            // 忽略，可能程序异常中断数据没有写完整
            log.warn("程序可能异常中断，索引数据不完整!!!");
        }
    }

    public void release() throws IOException{
        if(this.indexFileLock != null) {
            this.indexFileLock.release();
        }
    }
}
